package com.yomob.kylin.service;

import com.clearspring.analytics.util.Lists;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.UpdateManyModel;
import com.mongodb.client.model.UpdateOptions;
import com.yomob.kylin.client.MongoDBClient;
import com.yomob.kylin.facility.Consts;
import com.yomob.kylin.support.OptionSupport;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.storage.StorageLevel;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.stream.Collectors;

import static com.mongodb.client.model.Filters.eq;
import static org.apache.spark.sql.functions.col;

/**
 * Collect user payment action info,contains consume times,monetary
 */
public class PaymentService extends SimpleService {
    private static final Object[] COUNTERIDS = {"yomob_payment"};
    private static final Logger LOGGER = LoggerFactory.getLogger(PaymentService.class);
    private static final String COLLECTION_NAME = "daily_user_info";
    private MongoDBClient mongoDBClient;
    public PaymentService(MongoDBClient mongoDBClient) {
        this.mongoDBClient = mongoDBClient;
    }

    /**
     * Running Entry for payment service
     * @param input,spark data set during the start-day and end-day
     * @throws Exception
     */
    @Override
    public void execute(Dataset<Row> input) throws Exception {
        Dataset<Row> dataset = input.filter(col(Consts.Fields.COUNTERID).isin("yomob_payment")).select(
                Consts.Fields.UDID
                //todo
                //""//消费金额
        );
        dataset.persist(StorageLevel.fromString(OptionSupport.getStorageLevel().name()));
        List<Row> rowList = dataset.collectAsList();
        List<UpdateManyModel<Document>> models = Lists.newArrayList();
        models.addAll(rowList.stream().map(
                row -> this.convertPayment(row)
        ).collect(Collectors.toList()));
        mongoDBClient.saveToMongoBulk(models, COLLECTION_NAME);
        dataset.unpersist();
    }
    /**
     * Return this service deal counters type
     * @return String array
     */
    @Override
    public Object[] getCounterids() {
        return COUNTERIDS ;
    }

    /**
     * Private method,used to convert spark row data to mongo update command set,contain $set and $inc.
     * one row data means one user become payment-user,and inc user's consume times,and add total monetary.
     * @param row
     * @return mongodb UpdateManyModel<Document> for batch update
     */
    private UpdateManyModel<Document> convertPayment(Row row) {
        List<Bson> matches = Lists.newArrayList();
        Bson eq = eq(Consts.UserInfoBased.DEVICE_ID, row.getAs(Consts.Fields.UDID));
        matches.add(eq);
        Bson filter = Filters.and(matches);
        Document update = new Document();
        update.append(Consts.MongoCommand.SET,new Document(Consts.UserInfoBehavior.PAY_USER,true));
        update.append(Consts.MongoCommand.INC, new Document(Consts.UserInfoBehavior.CONSUME_TIMES, 1));
        //todo ,if mobile client can offer monetary price
        //update.append(Consts.MongoCommand.INC, new Document(Consts.UserInfoBehavior.CONSUME_TOTAL, Double.valueOf(row.getAs(1))));
        //update.append(Consts.MongoCommand.MAX, new Document(Consts.UserInfoBehavior.UNIT_PRICE_MAX, Double.valueOf(row.getAs(1))));

        UpdateOptions updateOptions = new UpdateOptions().upsert(true);
        return new UpdateManyModel<>(filter, update, updateOptions);
    }
}
